package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gmall.realtime.app.functions.DimSink;
import com.atguigu.gmall.realtime.app.functions.MyCustomSchema;
import com.atguigu.gmall.realtime.app.functions.TableProcessFunction;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.sun.xml.internal.bind.v2.TODO;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * @Description: 对业务数据进行处理, 处理方式 : 通过maxwall监测具体的业务数据库 gmall0609.
 * 如果库中某张表的数据或者新增删除了某张表,则直接通过maxwall抓取并发送到 ods_base_db_m
 * @Author: tiancy
 * @Create: 2021/11/26
 */

/*
    TODO ODS层数据分流再发送到DWD层的分流思路 :
        通过maxwall,监测MySQL中gmall0609库中表以及表中数据的变化,如果数据变化,则直接通过maxwall抓取放到`ods_base_db_m`这个主题中.
        问题 : 如果我业务库`gmall0609`,在我程序启动后,里面还会存在表的变化呢 ? 如果是这样,又该为当前新增的表划分成`dwd`还是`dim`呢 ?
        - 如何讲抓取到的变化数据分开呢,或者换一句话讲:如果区分是事实表`dwd层数据`还是`dim维度层数据`呢 ? 这里给出的解决思路 : 使用FlinkCDC,自己指定一个配置表`tableProcess`,表中指定表属于`dwd`还是`dim`.
        使用FlinkCDC将当前配置表的信息通过程序读取进入当前程序,作为一条流并将这条流作为广播流向下广播.
        - 再通过Flink环境读取`ods_base_db_m`中的数据作为主流. 通过 connect算子,将两条流进行合并,合并后再使用`tableProcess`中存的信息,对主流中数据进行分流.

 */

/*
    需要启动的服务 : MySQL --> maxwall --> Kafka(zk) --> 程序
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 设置流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
//        //TODO 2.检查点相关设置
//        //2.1 开启检查点
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        //2.2 设置检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        //2.3 设置取消job之后是否保留检查点
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //2.4 设置重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
//        //2.5 设置状态后端
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/dbCK"));
//        //2.6 设置操作hadoop的用户
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
        String topic = "ods_base_db_m";
        String groupId = "base_db_app_group";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);

        // 3、获取业务数据中 ods_base_db_m 主题 数据形成`主流`数据.
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        // 4、将Kafka中的json字符串转化为json对象.
        SingleOutputStreamOperator<JSONObject> jsonObject = kafkaDS.map(JSON::parseObject);
        /*
            原始数据格式 :
                {
                    "database":"gmall0609","xid":7887,
                    "data":{"tm_name":"香奈儿xxoo","logo_url":"/static/default.jpg","id":11},
                    "old":{"tm_name":"香奈儿"},
                    "commit":true,"type":"update","table":"base_trademark","ts":1637937231
                 }
         */
        // 5、将ods层数据进行校验并过滤数据,合法数据会被留在当前流中等待处理.
        SingleOutputStreamOperator<JSONObject> ETLSuccessDS = jsonObject.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        boolean flag =
                                jsonObject.getString("table") != null &&
                                        jsonObject.getString("table").length() > 0 &&
                                        jsonObject.getJSONObject("data") != null &&
                                        jsonObject.getString("data").length() > 3;
                        return flag;
                    }
                }
        );
        // ETLSuccessDS.print(">>>>>>>>>>>> ods_change_data : ");

        //TODO 6.使用FlinkCDC读取配置表数据

        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall0609_realtime")
                .tableList("gmall0609_realtime.table_process")
                .username("root")
                .password("123456")
                .startupOptions(StartupOptions.initial())
                /*
                   自定义反序列化器,用于将监测修改表中的数据 按照自定义格式输出
                   输出格式类型 :
                   {"database":"gmall0609_realtime","data":{"name":"Hello","id":4,"age":30},"type":"insert","table":"t_user"}
                 */
                .deserializer(new MyCustomSchema())
                .build();

        //6.2 读取配置表中的变化数据 封装为流
        DataStreamSource<String> mySQLDS = env.addSource(sourceFunction);
        // mySQLDS.print(">>>>>>");

        /*
            6.3 将tableProcess表中变化的流设置状态,并向下广播. 这里广播的原因 : 我们整体的程序设置的全局并行度为 4,也就是说整体程序在主流上,会并行的开四个流.
            每一条主流中处理的数据,都需要和`FlinkCDC`中的处理的配置表关联,并根据这个配置表中的配置项进行分流.
            先定义一个广播状态描述器,其中key -> 表名:操作 , value -> TableProcess对象.
         */

        MapStateDescriptor<String, TableProcess> tableProcessStates = new MapStateDescriptor<>("BroadcastStatus", Types.STRING, Types.POJO(TableProcess.class));
        BroadcastStream<String> broadcastDS = mySQLDS.broadcast(tableProcessStates);

        // TODO 7、将两个流使用算子 connect()进行合并.(一国两制的流.) 其中需要将`FlinkCDC`处理的`tableProcess`表中监测到修改的数据,使用广播的方式进行向下传播.从而实现双流操作.(将主流（业务数据流）和广播流(配置数据流) 进行连接)
        BroadcastConnectedStream<JSONObject, String> connectedStream = ETLSuccessDS.connect(broadcastDS);
        /*
            两个流经过合并后,使用process算子,算子内只能使用两种函数类 , keyedBroadcastProcessFunction 和 BroadcastProcessFunction.
            这里不涉及到使用keyBy,因此需要使用 BroadcastProcessFunction函数类.
            BroadcastProcessFunction<JSONObject, String, JSONObject> 三个泛型 , IN1、IN2、out.
            其中 主流中输入的是上面ETL产生的json对象,广播流中输入的是自定义反序列化器产生的String类型,输出为主流中经过筛选的数据,也就是 dwd层的数据.
            广播流的输入数据格式 : {"database":"gmall0609_realtime","data":{"name":"Hello","id":4,"age":30},"type":"insert","table":"t_user"}
            主流中输入的数据格式 :
                {
                    "database":"gmall0609","xid":7887,
                    "data":{"tm_name":"香奈儿xxoo","logo_url":"/static/default.jpg","id":11},
                    "old":{"tm_name":"香奈儿"},
                    "commit":true,"type":"update","table":"base_trademark","ts":1637937231
                 }
         */

        // 定义侧输出流标签,这里侧输出流最终存放的是 进入到HBase中的 维度数据 dimTag
        OutputTag<JSONObject> dimTag = new OutputTag<JSONObject>("dimTag") {
        };
        // 8、动态分流的结果
        SingleOutputStreamOperator<JSONObject> dynamicShuntResults = connectedStream.process(new TableProcessFunction(dimTag, tableProcessStates));
        //获取维度侧输出流数据
        DataStream<JSONObject> dimDS = dynamicShuntResults.getSideOutput(dimTag);
        // 主流数据,也就是事实表中的数据
        dynamicShuntResults.print(" >>>>>dwd层数据:<<<<<");
        // 侧输出流 : 维度表中的数据.
        dimDS.print("====dim层数据====");


        // 9、将事实表中的数据和业务表中的数据写入到Kafka或者HBase中
        /*
            TODO 何时创建维度表 ? 这个代码在哪写呢 ?  --> 动态分流时,读取配置表并将配置表中属性写入到广播状态时,来判断并创建HBase表结构 ==> TableProcessFunction.class 110行具体逻辑.
            先来判断侧输出流中的数据属于哪张表,并判断当前表是否存在,如果存在直接插入.如果不存在则先创建表再插入数据.

         */
        //TODO 10. 将维度侧输出流的数据写到phoenix表中
        dimDS.addSink(new DimSink());

        /*
            TODO 11、往Kafka中写入主流中`事实`数据,这里需要注意 :
             1、我们往Kafka中写入数据时.是一个生产者,并且每条数据写入的主题会不一致,并非一直往一个主题中写的,因此需要我们拿到当前数据发往的主题.
             2、还需要想办法实现精准一次性原则. ==>这里手动创建一个序列化过程,指定当前行数据如何发往下游的.
         */
        dynamicShuntResults.addSink(MyKafkaUtil.getKafkaByCustomSchema(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, @Nullable Long aLong) {
                // 通过当前行数据,获取将要发送的主题.
                String curDataTopic = jsonObject.getString("sink_table");
                return new ProducerRecord<byte[], byte[]>(curDataTopic, jsonObject.getJSONObject("data").toString().getBytes());
            }
        }));

        env.execute();
    }
}
